from pyspark.context import SparkContext

sc = SparkContext(master='local', appName='wordcount')
# 读取数据
data_rdd = sc.textFile('../../data/pages.txt')
# 整理数据
pages_rdd = data_rdd.map(lambda x: (x.split(',')[0], x.split(',')[1].split('|')))
# 缓存
pages_rdd.cache()
# 计算网页的数量
N = pages_rdd.count()

Q = 0.85

init_pr_rdd = pages_rdd.map(lambda x: (x[0], x[1], 1.0))

while True:
    # 4、将每个网页的pr平分给出链列表
    def avg_pr_func(pages):
        linked = pages[1]
        pr = pages[2]
        # 计算每个网页分给其他网页的pr值
        avg_pr = pr / len(linked)
        # 将pr值和网页关联
        link_pr = [(i, avg_pr) for i in linked]
        return link_pr


    avg_pr_rdd = init_pr_rdd.flatMap(avg_pr_func)
    # 5、计算新的PR值
    new_pr_rdd = avg_pr_rdd.reduceByKey(lambda x, y: x + y)
    # 增加阻尼系数
    new_pr_rdd = new_pr_rdd.map(lambda x: (x[0], (1 - Q) / N + Q * x[1]))
    # 6、关联获取出链列表
    join_rdd = new_pr_rdd.join(pages_rdd, numPartitions=1)
    # 7、整理格式
    new_pr_linked_rdd = join_rdd.map(lambda x: (x[0], x[1][1], x[1][0]))
    #整合数据，计算pr值和前一次pr的差值平均值
    new_pr_init_pr_rdd = new_pr_rdd.join(init_pr_rdd.map(lambda x: (x[0], x[2])))
    # 计算差值
    diff_pr = new_pr_init_pr_rdd.map(lambda x: abs(x[1][0] - x[1][1])).sum()
    # 计算平均值
    avg_diff_pr = diff_pr / N

    if avg_diff_pr < 0.01:
        new_pr_linked_rdd.saveAsTextFile('../../data/pages_linked.txt')
        break
    # 切换新的pr值进行下一次计算
    init_pr_rdd = new_pr_linked_rdd
    print('=' * 100)
